package cn.ac.iscas.service.transformer.dameng;

import cn.ac.iscas.base.utils.SpringUtil;
import cn.ac.iscas.base.utils.TypeConvertUtil;
import cn.ac.iscas.domain.flink.FlinkColumnType;
import cn.ac.iscas.domain.source.DamengSourceColumn;
import cn.ac.iscas.domain.source.FlinkSourceColumn;
import cn.ac.iscas.service.queue.dameng.DamengTableQueueService;
import cn.ac.iscas.service.queue.flink.FlinkTableQueueService;
import cn.ac.iscas.utils.DamengSqlUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @description:
 * @param:
 * 数据转换算子格式
 *  {
 *      //指定源列，只显示整型、浮点、字符串三种类型的列，传回列名和类型
 *      "src_column": {
 *          "column": "column1",
 *          "type": "STRING"
 *      },
 *      //目标列，只显示整型、浮点、字符串三种类型的列
 *      "target_column": {
 *          "column": "column2",
 *          "type": "Double"
 *      },
 *      //转换选项，以下几种情况
 *      //（1）整型转浮点，不出现选项下拉框，没有"options"
 *      //（2）浮点转整型，出现三个选项：ceil：向上取整，floor：向下取整，round：四舍五入（默认选中）
 *      //（3）字符串转整型，选项如（2）
 *      //（4）字符串转浮点，如（1）
 *      "options": {
 *          "mode": "ceil/floor/round"
 *      }
 *  }
 * @return:
 * @author LJian
 * @date: 2022/11/16 14:50
 */
public class DamengDataFormatTransformer {
    private final static String ID = "data_format";
    private final static String SRC_COLUMN_KEY = "src_column";
    private final static String SRC_COLUMN_TYPE = "src_column_type";
    private final static String COLUMN_NAME_KEY = "column";
    private final static String TARGET_COLUMN_KEY = "target_column";
    private final static String TARGET_COLUMN_TYPE = "target_column_type";
    private final static String COLUMN_TYPE_KEY = "type";
    private final static String OPTIONS_KEY = "options";
    private final static String MODE_KEY = "mode";
    private String jobId;
    private String componentId;
    private String frontTableName;
    private List<DamengSourceColumn> frontColumns;

    private String currentTableName;
    private List<DamengSourceColumn> currentColumns;

    public DamengDataFormatTransformer(String jobId, String taskId) {
        this.jobId = jobId;
        this.componentId = taskId;
    }

    public DamengDataFormatTransformer(String jobId, String taskId, String frontTableName, List<DamengSourceColumn> frontColumns) {
        this.jobId = jobId;
        this.componentId = taskId;
        this.frontTableName = frontTableName;
        this.frontColumns = frontColumns;
        this.currentColumns = frontColumns;
        this.currentTableName = String.format("\"%s\".\"%s_%s\"", jobId, ID, UUID.randomUUID().toString().substring(0,6));
    }


    public DamengDataFormatTransformer() {

    }


    public List<String> getSql(Map<String, Object> parameters) {
        List<String> sqls = new ArrayList<>();

        //Map<String, Object> nullFilParameter = parameters;
        String srcColumn = parameters.getOrDefault(SRC_COLUMN_KEY, "").toString();
        String srcColumnType = parameters.getOrDefault(SRC_COLUMN_TYPE, "").toString();
        String targetColumn = parameters.getOrDefault(TARGET_COLUMN_KEY, "").toString();
        String targetColumnType = parameters.getOrDefault(TARGET_COLUMN_TYPE, "").toString();

        String insertTempTableSql = DamengSqlUtils.getInsertOrderColumnSql(this.currentTableName, this.frontTableName, this.currentColumns);

        boolean isFrontTableContainsTargetColumn = this.currentColumns.stream()
                .anyMatch(column -> targetColumn.equals(column.getName()));
        if (!isFrontTableContainsTargetColumn) {
            DamengSourceColumn newColumn = new DamengSourceColumn();
            newColumn.setName(targetColumn);
            newColumn.setType(TypeConvertUtil.toDamengType(targetColumnType).toString());
            this.currentColumns.add(newColumn);
        }
        String createTempTableSql = DamengSqlUtils.getCreateTemporaryTableSql(this.currentTableName, this.currentColumns);

        String updateTempTableSql = String.format("UPDATE %s SET %s = CAST(%s AS %s)", this.currentTableName, targetColumn, srcColumn, targetColumnType);

        sqls.add(createTempTableSql);
        sqls.add(insertTempTableSql);
        sqls.add(updateTempTableSql);
        updateSource();
        return sqls;
    }

    public void updateSource() {
        DamengTableQueueService damengTableQueueService = SpringUtil.getBean(DamengTableQueueService.class);
        Map<String, List<DamengSourceColumn>> tableColumns = new ConcurrentHashMap<>(8);
        tableColumns.put(this.currentTableName, this.currentColumns);
        damengTableQueueService.put(this.jobId, this.componentId, tableColumns);
    }
}
